package com.zjw.socket.demo.starter.messageService.im;

import com.zjw.socket.demo.starter.core.ioc.IOCScope;
import com.zjw.socket.demo.starter.messageService.*;
import com.zjw.socket.demo.starter.messageService.service.IMessageFadeService;
import com.zjw.socket.demo.starter.messageService.service.impl.IMServiceConnect;
import com.zjw.socket.demo.starter.utils.LogUtils;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;

//demo
public class ImService {
  private static final String TAG = "NIOSelectorService";
  //连接池
  private final Map<Long, Channel> connectPool = new HashMap<>(32);
  private final Map<Channel, Long> connectPool2 = new HashMap<>(32);

  private ImServiceConfig config;

  private IMessageFadeService fadeService;

  @SuppressWarnings("all")
  public void start(ImServiceConfig config) throws Exception {
    this.config = config;

    //todo 临时这么解构
    fadeService = IOCScope.globalScope.getBean("SimpleMessageFadeService");

    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

    serverSocketChannel.bind(new InetSocketAddress(config.port));

    //设置为非阻塞
    serverSocketChannel.configureBlocking(false);

    //创建选择器
    Selector selector = Selector.open();

    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

    LogUtils.debug(TAG, "IM服务器启动成功 ");

    fadeService.imServiceInit();

    loop(serverSocketChannel, selector);
  }

  private void loop(ServerSocketChannel serverSocketChannel, Selector selector) throws Exception {
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    while (true) {
      //超时
      int select = selector.select(0);
      if (select == 0) {
        continue;
      }

      Set<SelectionKey> selectionKeys = selector.selectedKeys();
      Iterator<SelectionKey> selectionKeyIterator = selectionKeys.iterator();
      while (selectionKeyIterator.hasNext()) {
        SelectionKey key = selectionKeyIterator.next();
        //得到客户端通道，并将通道注册到选择器上，并且指定监听事件为OP_READ
        if (key.isAcceptable()) {
          SocketChannel channel = serverSocketChannel.accept();
          channel.configureBlocking(false);
          //监听读事件
          channel.register(selector, SelectionKey.OP_READ);
        }

        if (!key.isValid()) {
          SocketChannel channel = (SocketChannel) key.channel();
          channel.close();
          //断开连接的一般处理
          processConnectClose(channel);
          continue;
        }

        if (key.isReadable()) {
          SocketChannel channel = (SocketChannel) key.channel();

          if (!channel.isConnected()) {
            processConnectClose(channel);
            continue;
          }
          byteBuffer.clear();
          int len;
          ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
          try {
            while ((len = channel.read(byteBuffer)) > 0) {
              byteArrayOutputStream.write(byteBuffer.array(), 0, len);
            }

            if (len == -1) {
              LogUtils.debug(TAG, "连接关闭");
              channel.close();
              //close
              processConnectClose(channel);
              continue;
            }
          } catch (IOException e) {
            LogUtils.error(TAG, e);
            channel.close();
            processConnectClose(channel);
            continue;
          }

          byte[] bytes = byteArrayOutputStream.toByteArray();
          Message message;
          try {
            message = MessageParser.fromByte(bytes);
          } catch (Exception e) {
            LogUtils.error(TAG, e);
            continue;
          }
          if (!connectPool.containsKey(message.getFromId())) {
            connectPool.put(message.getFromId(), channel);
            connectPool2.put(channel, message.getFromId());
          }
          sendMessageToService(message);
        }
        selectionKeyIterator.remove();
      }
    }
  }

  private void processConnectClose(Channel socketChannel) {
    Long userId = connectPool2.remove(socketChannel);
    if (userId != null) {
      Message message = MessageCacheFactory.obtain();
      message.setFromId(userId);
      message.setTime(new Date().getTime());
      message.setType(SystemMessageType.USER_CONNECT_CLOSE);
      Channel channel = connectPool.remove(userId);
      if (channel != null) {
        try {
          channel.close();
        } catch (IOException e) {
          LogUtils.error(TAG, e);
        }
      }
      sendMessageToService(message);
    }
  }

  private void sendMessageToService(Message message) {
    message.setChanel(new IMServiceConnect(this, message.getFromId()));
    fadeService.transferMessageToService(message);
    MessageCacheFactory.recycle(message);
  }


  /**
   * 消息终端与此service连接才能进行发送
   */
  public boolean sendMessageToDestination(Message message) {
    long toId = message.getToId();
    Channel channel = connectPool.get(toId);
    if (channel == null) {
      LogUtils.error(TAG, toId + " not connect");
      return false;
    }

    onPreProcessMessageForWrite(message);

    ByteBuffer byteBuffer = MessageParser.toByteBuffer(message);
    if (channel instanceof SocketChannel) {
      try {
        ((SocketChannel) channel).write(byteBuffer);
        return true;
      } catch (IOException e) {
        LogUtils.error(TAG, toId + " " + e.getMessage());
        return false;
      }
    }

    if (channel instanceof DatagramChannel) {
      try {
        ((DatagramChannel) channel).write(byteBuffer);
        return true;
      } catch (IOException e) {
        LogUtils.error(TAG, toId + " " + e.getMessage());
        return false;
      }
    }
    return false;
  }

  private void onPreProcessMessageForWrite(Message message) {
    if (MessageProcessStrategy.hasFlag(message.getProcessStrategy(), MessageProcessStrategy.BODY_ENCRYPTION_MASK)) {
      //todo 消息加密
    }
  }

  public void disConnect(long uid) {
    Channel channel = connectPool.get(uid);
    if (channel != null) {
      try {
        channel.close();
      } catch (IOException e) {
        LogUtils.error(TAG, e);
      }
    }
  }
}
